"""
elio总框架。
"""
import abc
import asyncio
import traceback
import typing
from concurrent import futures  # 多线程包
from loguru import logger
from urllib.parse import parse_qs

import grpc  # type: ignore
import socketio  # type: ignore
from google.protobuf.message import DecodeError
from tornado import web

from ..tornado import CorsProxyHandler, FileUploadHandler
from .utils import Any, Int64Value, get_local_ip, to_any
from ..proto import ElegantApiServicer, Interaction, add_ElegantApiServicer_to_server  # type: ignore

_thread_pool_executor = futures.ThreadPoolExecutor(max_workers=10)
_process_pool_executor = futures.ProcessPoolExecutor()
_loop = asyncio.new_event_loop()
asyncio.set_event_loop(_loop)
_config: dict = {}


def call(coro: typing.Coroutine):
    """
    在当前进程中运行一个携程，不适用事件循环，这意味着和调用普通函数一样会阻塞当前线程。
    :param coro: 携程对象。
    """
    ret = None
    try:
        while True:
            ret = coro.send(None)
    except StopIteration:
        pass
    return ret


def dispatch(coro: typing.Coroutine):
    """
    在当前进程中使用事件循环运行一个携程，但不等待他的结果。
    :param coro: 携程对象。
    """
    _loop.create_task(coro)


def get_event_loop() -> asyncio.AbstractEventLoop:
    """
    获取elio框架使用的事件循环。
    """
    return _loop


def get_thread_pool_executor() -> futures.ThreadPoolExecutor:
    """
    获取框架使用的线程池。
    """
    return _thread_pool_executor


def get_process_pool_executor() -> futures.ProcessPoolExecutor:
    """
    获取框架使用的进程池。
    """
    return _process_pool_executor


def run_in_process_pool(func, *args):
    """
    运行一个CPU计算密集的任务（通常耗时不小于1秒），把func函数推送到进程池中运行以便于不阻塞elio框架的事件循环。
    :param func: 一个非异步的函数对象
    :param args: 函数要用到的参数
    """
    return _loop.run_in_executor(_process_pool_executor, func, *args)


def shutdown():
    """
    关闭整个框架，用到的所有资源也将被释放。
    """
    _process_pool_executor.shutdown()
    _thread_pool_executor.shutdown()


_sessions: typing.Dict[int, typing.Any] = {}
SessionType = typing.TypeVar("SessionType", bound='Session')


class Session(metaclass=abc.ABCMeta):
    """
    会话的抽象类，属于框架的最低级API，是每一个elegantio框架组件的基础。
    """

    def __init__(self, context):
        """
        会话对象初始化。
        """
        cygrpc = getattr(grpc, '_cython').cygrpc
        if not (isinstance(context, getattr(cygrpc, '_ServicerContext')) or isinstance(context, dict)):
            raise RuntimeError('%s instance cannot be new by the user.' % self.__class__.__name__)
        self._active = False
        self._context = context
        self._queue = asyncio.Queue(10)

    def __del__(self):
        """
        会话对象被回收，在此释放内存。
        """
        if hasattr(self, '_active') and self._active:
            dispatch(self.on_destroy())
            del self._active
        if hasattr(self, '_queue'):
            del self._queue
        if hasattr(self, '_context'):
            del self._context

    async def __aiter__(self):
        """
        获取发送队列的迭代器，此方法是一个异步生成器。
        """
        while hasattr(self, '_queue'):
            yield await self._queue.get()
            self._queue.task_done()

    async def send(self, data: Interaction):
        """
        发送交互数据。
        """
        if not hasattr(self, '_active') or not self._active:
            return
        if hasattr(self, '_queue'):
            return await self._queue.put(data)

    async def _receive_message(self, message_receiver):
        """
        循环接收会话的数据并处理，直到会话结束。
        此方法为内部方法。
        """
        try:
            if isinstance(message_receiver, Interaction):
                await self.on_receive(message_receiver)
                return
            async for rec in message_receiver:
                await self.on_receive(rec)
        except DecodeError as e:
            await self.on_error(e)
            logger.trace(traceback.format_exc())

    async def destroy(self):
        """
        销毁当前会话。
        """
        if hasattr(self, '_active') and self._active:
            await self.on_destroy()
            del self._active

    @staticmethod
    def find_sessions_by_client_id(client_id: str) -> set[SessionType]:
        """
        通过客户端ID查找所有会话的实例对象并返回。
        :param client_id: 客户端ID。
        """
        return {i for i in _sessions.values() if i.client_id == client_id}

    @staticmethod
    def find_session_by_id(session_id: int) -> typing.Optional[SessionType]:
        """
        通过ID查找一个会话的实例对象并返回。
        :param session_id: 会话的ID。
        """
        if session_id in _sessions:
            return _sessions[session_id]
        return None

    @classmethod
    @abc.abstractmethod
    async def create(
            cls: typing.Type[SessionType], parent: typing.Any, *args: typing.Any, **kwargs: typing.Dict
    ) -> SessionType:
        """
        调用此函数将创建一个通信会话并返回，用于和设备端的组件通信。
        :pram cls: 要创建基于Session会话的实现类；
        :param parent: 创建的来源对象。
        """
        ...

    @abc.abstractmethod
    async def on_create(self):
        """
        如果当前会话已创建，此方法将被自动调用。
        """
        self._active = True
        if self.is_debug:
            logger.debug('%s: onCreated.' % self.__class__.__name__)

    @abc.abstractmethod
    async def on_destroy(self):
        """
        如果当前会话销毁，此方法将被自动调用。
        """
        if not hasattr(self, '_active') or not self._active:
            return
        if self.is_debug:
            logger.debug('%s: onDestroyed.' % self.__class__.__name__)
        self._active = False

    @abc.abstractmethod
    async def on_error(self, exception: typing.Union[Exception, Any], **kwargs):
        """
        如果当前会话发生了一个错误，此方法将被自动调用。
        :param exception: 异常信息。
        """
        if self.is_debug:
            logger.debug('%s error: %s' % (self.__class__.__name__, exception))

    @abc.abstractmethod
    async def on_receive(self, msg):
        """
        如果当前会话接收到一条交互消息，此方法将被自动调用。
        """
        ...

    @property
    def reference(self) -> Any:
        """
        获取本会话的引用，把Session的ID转换成protobuf的对象从而进行传输。
        """
        return to_any(self.id, Int64Value)

    def _get_context_data(self, key: str) -> typing.Any:
        if isinstance(self._context, dict):
            query = parse_qs(self._context['QUERY_STRING'])
            return query[key][0] if key in query else None
        for k, v in self._context.invocation_metadata():  # type: ignore
            if k == key:
                return v
        return None

    @property
    def is_debug(self) -> bool:
        """
        是否为调试模式。
        """
        return _config['debug'] if 'debug' in _config else False

    @property
    def is_closed(self) -> bool:
        """
        判断当前会话是否已经关闭。
        """
        if hasattr(self, '_context'):
            return self._context.cancelled()  # type: ignore
        return True

    @property
    def id(self) -> int:
        """
        获取当前会话的ID。
        """
        return id(self)

    @property
    def client_id(self) -> typing.Optional[str]:
        """
        获取设备的客户ID。
        """
        return self._get_context_data('client-id')

    @property
    def client_ip(self) -> str:
        """
        获取客户端的IP地址。
        """
        if isinstance(self._context, dict):
            host = self._context['REMOTE_ADDR']
            port = self._context['REMOTE_PORT']
            return f'IPV4:{host}:{port}'
        return self._context.peer()  # type: ignore

    @property
    def user_agent(self) -> typing.Optional[str]:
        """
        获取设备的用户代理。
        """
        if isinstance(self._context, dict):
            return self._context['HTTP_USER_AGENT']
        return self._get_context_data('user-agent')

    @property
    def platform(self) -> typing.Optional[str]:
        """
        获取设备的操作系统平台。
        """
        return self._get_context_data('platform')


servicer = ElegantApiServicer()
_sessionClasses = {}
_T1 = typing.TypeVar("_T1", bound=typing.Type)


def session_received(cls: _T1) -> _T1:
    """
    会话流接受装饰器，将此方法标注在组件类上，可以实现该组件与设备端组件通信。
    如果设备端的组件从grpc协议发起了调用，此装饰器将通过流传输方式把数据传送到设备端。
    如果设备端使用websocket协议调用，此方法会把任务交给socket-io框架处理。
    """
    if not hasattr(cls, 'name'):
        raise RuntimeError('%s has not setted valid `name` field, and it will not be instantiated.' % cls.__name__)
    _sessionClasses[cls.name] = cls

    async def inject(message_receiver, context):
        """
        动态把处理函数注入grpcio。
        """
        global _sessions
        try:
            session = cls(context)
        except RuntimeError:
            logger.trace(traceback.format_exc())
            return
        session_id = session.id
        _sessions[session_id] = session
        dispatch(getattr(session, '_receive_message')(message_receiver))
        try:
            async for ret in session:
                if ret:
                    yield ret
        except Exception as e:
            dispatch(session.on_error(e))
            logger.trace(traceback.format_exc())
        finally:
            try:
                if session_id in _sessions:
                    del _sessions[session_id]
                await session.destroy()
                del session
            except (KeyError, TypeError, AttributeError, RuntimeError, NameError):
                logger.trace(traceback.format_exc())

    setattr(servicer, cls.name, inject)
    return cls


_T2 = typing.TypeVar("_T2", bound=typing.Type)


def session_requested(cls: _T2) -> _T2:
    """
    会话请求装饰器，将此方法标注在组件类上，可以实现该组件与设备端组件通信。
    如果设备端的组件使用grpc协议发起了调用，此装饰器将发送一个响应结果。
    如果设备端使用websocket协议调用，此方法会把任务交给socket-io框架处理。
    """
    if not hasattr(cls, 'name'):
        raise RuntimeError('%s has not setted valid `name` field, and it will not be instantiated.' % cls.__name__)
    _sessionClasses[cls.name] = cls

    async def inject(message_receiver, context):
        """
        请求处理函数动态注入。
        """
        global _sessions
        try:
            session = cls(context)
        except RuntimeError:
            logger.trace(traceback.format_exc())
            return
        session_id = session.id
        _sessions[session_id] = session
        await getattr(session, '_receive_message')(message_receiver)
        res = None
        try:
            async for ret in session:
                res = ret
        except Exception as e:
            dispatch(session.on_error(e))
            logger.trace(traceback.format_exc())
        finally:
            if session_id in _sessions:
                del _sessions[session_id]
            await session.destroy()
            del session
        return res

    setattr(servicer, cls.name, inject)
    return cls


_tornadoApp: typing.Optional[web.Application] = None


def cors_proxy_serve(route: str) -> str:
    """
    注册一个跨域代理服务，用于解决web前端无法访问某些响应头无'Access-Control-Allow-Origin'的第三方资源url，返回注册后的baseUrl。
    :param route: 路由名称
    """
    base_url = f'http://{get_local_ip()}:8001{route}'

    def cb():
        """
        内部使用。
        """
        if not _tornadoApp:
            return
        _tornadoApp.add_handlers('.*', [(route, CorsProxyHandler,)])
        logger.success(f'Served cors proxy on URL: {base_url}')

    _loop.call_later(1, cb)
    return base_url


def file_upload_serve(path: str, route: str) -> str:
    """
    注册一个文件上传服务。
    :param path: 文件夹路径，要把接收的文件存放在哪里。
    :param route: 路由名称
    """
    base_url = f'http://{get_local_ip()}:8001{route}'

    def cb():
        """
        内部使用。
        """
        if not _tornadoApp:
            return
        _tornadoApp.add_handlers('.*', [(route, FileUploadHandler, {
            'path': path
        },)])
        logger.success(f'Served file upload on URL: {base_url}')

    _loop.call_later(1, cb)
    return base_url


def static_file_serve(path: str, route: str) -> str:
    """
    注册一个用于分发静态文件的服务，返回注册后的baseUrl。
    :param path: 文件夹路径
    :param route: 路由名称
    """
    base_url = f'http://{get_local_ip()}:8001{route}'

    def cb():
        """
        内部使用。
        """
        if not _tornadoApp:
            return
        _tornadoApp.add_handlers('.*', [(route + '/(.*)', web.StaticFileHandler, {
            'path': path
        })])
        logger.success(f'Served static files on URL: {base_url}')

    _loop.call_later(1, cb)
    return base_url


async def _register_sio_handlers(sio, cls):
    namespace = '/' + cls.name

    async def send(sid: str):
        """
        socketio的发送处理。
        """
        try:
            async with sio.session(sid, namespace) as s:
                session = s['session']
                try:
                    async for ret in session:
                        await sio.emit('call', ret.SerializeToString(), namespace=namespace, to=sid)
                except Exception as e:
                    dispatch(session.on_error(e))
                    logger.trace(traceback.format_exc())
        except KeyError:
            pass

    async def recv(session: Session, msg: bytes):
        """
        socketio的接收处理。
        """
        try:
            action = Interaction.FromString(msg)
            await session.on_receive(action)
        except DecodeError as e:
            await session.on_error(e)
            logger.trace(traceback.format_exc())

    @sio.on('connect', namespace=namespace)
    async def connect(sid, environ):
        """
        socketio的连接处理。
        """
        async with sio.session(sid, namespace) as s:
            try:
                session = cls(environ)
                s['session'] = session
            except RuntimeError:
                logger.trace(traceback.format_exc())
                return
            _sessions[session.id] = session
            dispatch(send(sid))

    @sio.on('disconnect', namespace=namespace)
    async def disconnect(sid):
        """
        socketio的断开连接处理。
        """
        async with sio.session(sid, namespace) as s:
            if 'session' not in s:
                return
            try:
                if s['session'].id in _sessions:
                    del _sessions[s['session'].id]
            except NameError:
                pass
            await s['session'].destroy()
            del s['session']

    @sio.on('call', namespace=namespace)
    async def call(sid, msg):
        """
        socketio调用方法的处理。
        """
        async with sio.session(sid, namespace) as s:
            if 'session' not in s:
                return
            session = s['session']
            dispatch(recv(session, msg))


async def register_to_server(
        grpcio_server: grpc.Server, socketio_server: socketio.Server, tornado_application: web.Application, **kwargs
):
    """
    注册服务器，将grpcio和socketio的服务器对象同时与elegantio框架的核心绑定。
    这意味着elegantio框架同时实现了多种传输协议（grpcio、socketio）。
    :param grpcio_server: GRPC的服务器对象。
    :param socketio_server: socketio的服务器对象。
    :param tornado_application: tornado web应用对象。
    :param kwargs: 整个框架所用的所有配置参数。
    """
    global _config, _tornadoApp
    _config = kwargs
    _tornadoApp = tornado_application
    add_ElegantApiServicer_to_server(servicer=servicer, server=grpcio_server)
    for _, cls in _sessionClasses.items():
        await _register_sio_handlers(socketio_server, cls)


async def unregister_from_server():
    """
    注销服务器。
    """
    global servicer, _sessions
    for k in list(_sessions):
        await _sessions[k].destroy()
        del _sessions[k]
    del servicer, _sessions
